package spring.cloud.tasks.tasks_manager.task_item_assign;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.TaskType;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.TaskTtemAssignNodePaths;
import spring.cloud.tasks.common.task_item_allot.Executor;
import spring.cloud.tasks.common.task_item_allot.TaskItem;
import spring.cloud.tasks.tasks_manager.schedule.ScheduleManager;
import spring.cloud.tasks.tasks_manager.schedule.TaskIdListContext;
import spring.cloud.tasks.tasks_manager.schedule.TaskInstanceInfo;
import spring.cloud.tasks.tasks_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorLostTraffic;
import spring.cloud.tasks.tasks_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorOffline;
import spring.cloud.tasks.tasks_manager.task_item_assign.pick_up.executor.PickUpWhenExecutorRecoverTraffic;
import spring.cloud.tasks.tasks_manager.task_item_assign.pick_up.task.PickUpWhenTaskDisable;
import spring.cloud.tasks.tasks_manager.waitingPreAssignTasks.SlidingTimeTaskBucketQueueService;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * 针对每个任务实例+stage+任务分片项都需要提前在ZK上建立对应的节点，如果没有建立成功那么则不允许进行预分配调度。
 * 具体执行过程中通过分片项节点进行续期以保证状态正确更新。
 * 分片信息状态同步，以及stage整体状态汇总。如果超时了那么下一stage不进行查询，用户需要自己排查问题。
 */


@Slf4j
@Setter
@Getter
public class TaskItemAssign {

    protected ScheduleManager scheduleManager;

    public TaskItemAssign(ScheduleManager scheduleManager) {
        this.scheduleManager = scheduleManager;
    }

    /**
     * <pre>
     * 1根据enableTaskIdList计算出最近需要进行执行的任务。如果任务分配了但还没有到执行的时间 那么可以进行重新分配。
     * 2历史enableTaskIdList任务只要没有到达最大的timeoutDateTime那么则允许在任务分配列表上面，虽然该任务不进行分配，但是有可能任务还在执行，这些任务即使执行失败也不修改执行计划。
     * 3节点上线只能等任务下次分配。
     * 4节点下线可强制进行一次任务分配。这样做需要任务具备一个属性:下线的时候任务可以立即被重新分配（一天只执行一次 或者 有限的次数 那么允许重新分配，否则任务会被丢失执行）。
     * </pre>
     */
    public void taskItemAssign() throws Exception {
        TaskItemContainer taskItemContainer = scheduleManager.taskItemContainer;
        ExecutorTaskItemListContext assignedExecutorTaskItemListContext = taskItemContainer.assignedExecutorTaskItemListContext;
        /**
         * 正在调度分配的任务分片容器
         */
        ExecutorTaskItemListContext assigningExecutorTaskItemListContext = new ExecutorTaskItemListContext();
        List<Executor> executorList = assigningExecutorTaskItemListContext.getExecutorList();
        SlidingTimeTaskBucketQueueService slidingTimeTaskBucketQueueService = scheduleManager.slidingTimeTaskBucketQueueService;

        Curator curator = scheduleManager.curator;
        log.info("taskItemAssign start");
        try {
            List<Executor> oldOnlineExecutorList = assignedExecutorTaskItemListContext.getExecutorList();
            executorList.addAll(TaskItemAssignUtils.copyExecutorList(oldOnlineExecutorList));
            TaskIdListContext taskIdListContext = scheduleManager.taskIdListContext;
            taskIdListContext.refresh(curator);
            TaskIdListContext.TaskIdLists taskIdLists = taskIdListContext.taskIdLists;
            List<String> lastTaskIdList = taskIdLists.taskIdList;
            List<String> lastEnableTaskIdList = taskIdLists.enableTaskIdList;
            List<String> lastPauseTaskIdList = taskIdLists.pauseTaskIdList;
            //
            List<Executor> lastOnlineExecutorList = TaskItemAssignUtils.getLatestExecutorList(curator);
            List<String> lastOnlineExecutorIdList = TaskItemAssignUtils.getExecutorIdList(lastOnlineExecutorList);
            List<Executor> lastOnlineOwnTrafficExecutorList = TaskItemAssignUtils.getIsNotFlowControlExecutorList(lastOnlineExecutorList);
            List<String> lastOnlineOwnTrafficExecutorIdList = TaskItemAssignUtils.getIsNotFlowControlExecutorIdList(lastOnlineExecutorList);
            List<Executor> lastOnlineDoNotOwnTrafficExecutorList = TaskItemAssignUtils.getFlowControlExecutorList(lastOnlineExecutorList);
            List<String> lastOnlineDoNotOwnTrafficExecutorIdList = TaskItemAssignUtils.getFlowControlExecutorIdList(lastOnlineExecutorList);
            //
            TaskItemAssignPresentSituation taskItemAssignPresentSituation = new TaskItemAssignPresentSituation();
            taskItemAssignPresentSituation.oldOnlineExecutorList = executorList;//old对象
            //taskItemAssignPresentSituation.newOwnTrafficOnlineExecutorList = newOwnTrafficOnlineExecutorList;
            //taskItemAssignPresentSituation.newDoNotOwnTrafficOnlineExecutorList = newDoNotOwnTrafficOnlineExecutorList;
            //
            taskItemAssignPresentSituation.lastTaskIdList = lastTaskIdList;
            taskItemAssignPresentSituation.lastEnableTaskIdList = lastEnableTaskIdList;
            //
            taskItemAssignPresentSituation.lastOnlineExecutorIdList = lastOnlineExecutorIdList;
            taskItemAssignPresentSituation.lastOnlineOwnTrafficExecutorIdList = lastOnlineOwnTrafficExecutorIdList;
            taskItemAssignPresentSituation.lastOnlineDoNotOwnTrafficExecutorIdList = lastOnlineDoNotOwnTrafficExecutorIdList;
            //
            TaskItemAssignFixPoints taskItemAssignFixPoints = TaskItemAssignUtils.getTaskItemAssignFixPoints(taskItemAssignPresentSituation);
            List<String> taskIdList = taskItemAssignFixPoints.getTaskIdList();
            List<String> disableTaskIdList = taskItemAssignFixPoints.getDisableTaskIdList();//优先处理[1]
            List<String> enableTaskIdList = taskItemAssignFixPoints.getEnableTaskIdList();
            List<String> offlineExecutorIdList = taskItemAssignFixPoints.getOfflineExecutorIdList();
            List<String> onlineExecutorIdList = taskItemAssignFixPoints.getOnlineExecutorIdList();//按照新的算
            List<String> onlineDoNotOwnTrafficExecutorIdList = taskItemAssignFixPoints.getOnlineDoNotOwnTrafficExecutorIdList();//优先处理[2]
            List<String> onlineOwnTrafficExecutorIdList = taskItemAssignFixPoints.getOnlineOwnTrafficExecutorIdList();
            //更新执行器的支持的任务list
            TaskItemAssignUtils.updateIsOwnTrafficAndSupportedTaskIdList(executorList, lastOnlineExecutorList);
            //在上个版本的任务分配基础上新增新上线的执行器
            TaskItemAssignUtils.addNewOnlineExecutorList(executorList, lastOnlineExecutorList);
            //重新最新的在线以及在线拥有流量的执行器信息
            List<Executor> ownTrafficOnlineExecutorList = TaskItemAssignUtils.getIsNotFlowControlExecutorList(executorList);
            List<TaskItem> pickUpEdTaskItemList = new ArrayList<TaskItem>();
            List<TaskItem> discardTaskItemList = new ArrayList<TaskItem>();
            //nullPreferExecutorSettings 只是占位空对象
            PreferExecutorSettings nullPreferExecutorSettings = new PreferExecutorSettings();
            nullPreferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap = null;
            nullPreferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap = null;
            nullPreferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap = null;
            //先清理超时的(执行很久的任务)
            long nowDateTime = System.currentTimeMillis();
            for (Executor executor : executorList) {
                List<TaskItem> taskItemList = executor.getTaskItemList();
                Iterator<TaskItem> iterator = taskItemList.iterator();
                while (iterator.hasNext()) {
                    TaskItem taskItem = iterator.next();
                    String taskType = taskItem.getTaskType();
                    if ((TaskType.SIMPLE_SLICING.getTaskType().equals(taskType))) {
                        if (taskItem.getTimeoutDateTime().getTime() <= nowDateTime) {
                            iterator.remove();
                            continue;
                        }
                        //像这种超时 应该算作什么状态。
                    } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType) || TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType)) {
                        //没有维护的批次任务超时清理
                        if (taskItem.getTimeoutDateTime().getTime() <= nowDateTime) {
                            iterator.remove();
                            continue;
                        }
                    } else {
                        throw new IllegalArgumentException();
                    }
                }
            }
            //disableTaskIdList
            for (String disableTaskId : disableTaskIdList) {
                PickUpWhenTaskDisable pickUp = new PickUpWhenTaskDisable();
                pickUp.setTaskId(disableTaskId);
                pickUp.pickUp(curator,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        executorList,//
                        ownTrafficOnlineExecutorList,//
                        nullPreferExecutorSettings);
            }
            //offlineExecutorIdList,有的健康任务需要被放回
            for (String offlineExecutorId : offlineExecutorIdList) {
                PickUpWhenExecutorOffline pickUp = new PickUpWhenExecutorOffline();
                pickUp.setExecutorId(offlineExecutorId);
                pickUp.pickUp(curator,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        executorList,//
                        ownTrafficOnlineExecutorList,//
                        nullPreferExecutorSettings);
            }
            //onlineDoNotOwnTrafficExecutorIdList,有的健康任务需要被放回
            for (String onlineDoNotOwnTrafficExecutorId : onlineDoNotOwnTrafficExecutorIdList) {
                PickUpWhenExecutorLostTraffic pickUp = new PickUpWhenExecutorLostTraffic();
                pickUp.setExecutorId(onlineDoNotOwnTrafficExecutorId);
                pickUp.pickUp(curator,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        executorList,//
                        ownTrafficOnlineExecutorList,//
                        nullPreferExecutorSettings);
            }
            //摘取Executor不支持的任务
            for (Executor executor : executorList) {
                List<TaskItem> taskItemList = executor.getTaskItemList();
                Iterator<TaskItem> iterator = taskItemList.iterator();
                while (iterator.hasNext()) {
                    TaskItem taskItem = iterator.next();
                    if (!executor.getTaskIdList().contains(taskItem.getTaskId())) {
                        executor.setTotalLoadLevel(executor.getTotalLoadLevel() - taskItem.getTaskItemLoad());
                        iterator.remove();
                        pickUpEdTaskItemList.add(taskItem);
                    }
                }
            }
            //根据现有的任务实例，计算出，需要删除的任务实例，需要任务阶段创建或者清理新的任务分片列表
            //互相对照。
            List<TaskInstanceInfo> simpleTaskStateFullHasAssignedTaskInstanceInfoList = TaskItemAssignUtils.getSimpleTaskStateFullHasAssignedTaskInstanceInfoList(scheduleManager);
            List<TaskInstanceInfo> simpleTaskStateLessHasAssignedTaskInstanceInfoList = TaskItemAssignUtils.getSimpleTaskStateLessHasAssignedTaskInstanceInfoList(scheduleManager);
            List<TaskInstanceInfo> batchTaskHasAssignedTaskInstanceInfoList = TaskItemAssignUtils.getBatchTaskHasAssignedTaskInstanceInfoList(scheduleManager);
            //还活跃的任务实例id list
            List<String> taskInstanceIdList = new ArrayList<>();
            for (TaskInstanceInfo taskInstanceInfo : simpleTaskStateFullHasAssignedTaskInstanceInfoList) {
                taskInstanceIdList.add(taskInstanceInfo.getTaskInstanceId());
            }
            for (TaskInstanceInfo taskInstanceInfo : simpleTaskStateLessHasAssignedTaskInstanceInfoList) {
                taskInstanceIdList.add(taskInstanceInfo.getTaskInstanceId());
            }
            for (TaskInstanceInfo taskInstanceInfo : batchTaskHasAssignedTaskInstanceInfoList) {
                taskInstanceIdList.add(taskInstanceInfo.getTaskInstanceId());
            }
            {
                //删除没有必要的实例id
                for (Executor executor : executorList) {
                    List<TaskItem> taskItemList = executor.getTaskItemList();
                    Iterator<TaskItem> iterator = taskItemList.iterator();
                    while (iterator.hasNext()) {
                        TaskItem taskItem = iterator.next();
                        if (!taskInstanceIdList.contains(taskItem.getTaskInstanceId())) {
                            executor.setTotalLoadLevel(executor.getTotalLoadLevel() - taskItem.getTaskItemLoad());
                            iterator.remove();
                            discardTaskItemList.add(taskItem);
                        }
                    }
                }
                Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
                while (iterator.hasNext()) {
                    TaskItem taskItem = iterator.next();
                    if (!taskInstanceIdList.contains(taskItem.getTaskInstanceId())) {
                        iterator.remove();
                        discardTaskItemList.add(taskItem);
                    }
                }
            }
            //进行实例信息阶段分片信息调整。
            //简单任务分片，啥都不用做。
            //批次任务实例的阶段分片列表，只要没有到超时时间，该分片预分配，可以无限次重新分配。直到成功。
            for (TaskInstanceInfo taskInstanceInfo : batchTaskHasAssignedTaskInstanceInfoList) {
                TaskItemAssignUtils.rebuildBatchTaskNextTaskItemList(taskInstanceInfo);
                if (taskInstanceInfo.getRebuildAssignTaskItemPlanCreate() == null) {
                    continue;// 什么也不做
                } else {
                    {
                        //删除没有必要的实例id
                        for (Executor executor : executorList) {
                            List<TaskItem> taskItemList = executor.getTaskItemList();
                            Iterator<TaskItem> iterator = taskItemList.iterator();
                            while (iterator.hasNext()) {
                                TaskItem taskItem = iterator.next();
                                if (taskItem.getTaskInstanceId().equals(taskInstanceInfo.getTaskInstanceId())) {
                                    executor.setTotalLoadLevel(executor.getTotalLoadLevel() - taskItem.getTaskItemLoad());
                                    iterator.remove();
                                    discardTaskItemList.add(taskItem);
                                }
                            }
                        }
                        Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
                        while (iterator.hasNext()) {
                            TaskItem taskItem = iterator.next();
                            if (taskItem.getTaskInstanceId().equals(taskInstanceInfo.getTaskInstanceId())) {
                                iterator.remove();
                                discardTaskItemList.add(taskItem);
                            }
                        }
                    }
                    if (taskInstanceInfo.getRebuildAssignTaskItemPlanCreate()) {
                        pickUpEdTaskItemList.addAll(taskInstanceInfo.getTaskItemList());
                    }
                }
            }

            //任务启用,到了时间点就进行分配，根据waitedAssignTaskItemList计算出最近可以进行资源分配的任务列表
            List<TaskInstanceInfo> waitedAssignTaskInstanceInfoList = new ArrayList<>();
            List<TaskItem> waitedAssignTaskItemList = TaskItemAssignUtils.getWaitedAssignTaskItemList(//
                    scheduleManager,//
                    curator,//
                    slidingTimeTaskBucketQueueService,
                    enableTaskIdList,
                    waitedAssignTaskInstanceInfoList
            );

            //
            pickUpEdTaskItemList.addAll(waitedAssignTaskItemList);
            //剩下的是有效任务 以及在线执行器(可能有的执行器没有分配流量通过)
            PreferExecutorSettings preferExecutorSettings = new PreferExecutorSettings();
            for (String taskId : enableTaskIdList) {
                preferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap.put(taskId, TaskItemAssignUtils.preferExecutorIdListIsConfigured(curator, taskId));
                preferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap.put(taskId, TaskItemAssignUtils.getPreferExecutorIdListConfigured(curator, taskId, onlineExecutorIdList));
                preferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap.put(taskId, TaskItemAssignUtils.getSupportedTaskIdOwnTrafficOnlineExecutorList(ownTrafficOnlineExecutorList, taskId));
            }
            assigningExecutorTaskItemListContext.preferExecutorSettings = preferExecutorSettings;
            assigningExecutorTaskItemListContext.enableTaskIdList = enableTaskIdList;
            //将配置没有优先执行器list的任务或者不在优先执行器list执行器节点清除(非优选节点可以不用忙进行切换)。如果优选节点都挂了，那么任务则不进行分配。
            TaskItemAssignUtils.fixPreferOnlineExecutorList(enableTaskIdList, preferExecutorSettings, pickUpEdTaskItemList, ownTrafficOnlineExecutorList);
            //检查哪些执行器拥有流量但是没有进行任务分配,然后进行执行器恢复流量
            List<Executor> onlineOwnTrafficButNotAssignedExecutorList = TaskItemAssignUtils.getOnlineOwnTrafficButNotAssignedExecutorList(executorList);
            for (Executor executor : onlineOwnTrafficButNotAssignedExecutorList) {
                PickUpWhenExecutorRecoverTraffic pickUp = new PickUpWhenExecutorRecoverTraffic();
                pickUp.setExecutorId(executor.getExecutorId());
                pickUp.pickUp(curator,//
                        taskIdList,//
                        enableTaskIdList,//
                        pickUpEdTaskItemList,//
                        discardTaskItemList,//
                        executorList,//
                        ownTrafficOnlineExecutorList,//
                        preferExecutorSettings);
            }
            //等流量分配均匀后再进行下次任务调度分配  这里的逻辑可能是错误的 因为有的分配不需要均衡，可能还需要丢弃摘取一次。比如批次任务，状态阶段变更。
            boolean pickUpEd = pickUpEdTaskItemList.size() > 0;
            if (pickUpEd) {
                // 放回
                putBack(pickUpEdTaskItemList, executorList, preferExecutorSettings
                );
                // 持久化
                List<String> executorIdList = new ArrayList<>();
                for (Executor executor : assigningExecutorTaskItemListContext.getExecutorList()) {
                    executorIdList.add(executor.getExecutorId());
                }
                assigningExecutorTaskItemListContext.setExecutorIdList(executorIdList);
                taskItemContainer.assignedExecutorTaskItemListContext = assigningExecutorTaskItemListContext;
                assignedExecutorTaskItemListContext.getExecutorList().clear();
                assignedExecutorTaskItemListContext = null;
            }
            //通知ZK分配信息改变
            curator.merge(TaskTtemAssignNodePaths.getTaskItemNodesTasksVersionNodePath(), (assigningExecutorTaskItemListContext.getVersion() + ""));
            TaskItemAssignUtils.addWaitedAssignTaskItemList(scheduleManager, waitedAssignTaskInstanceInfoList);
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
            scheduleManager.setNeedTaskItemReAssign(false);
        } finally {
        }
        log.info("taskItemAssign end");
    }


    /**
     * 放回算法：拿取Shard，放进负荷最小的executor
     */
    public static void putBack(List<TaskItem> pickUpEdTaskItemList, List<Executor> onlineExecutorList, PreferExecutorSettings preferExecutorSettings) throws Exception {
        if (onlineExecutorList.isEmpty()) {
            log.warn("上线执行器列表为空,不进行放回");
            return;
        }
        TaskItemAssignUtils.sortTaskItemList(pickUpEdTaskItemList);
        // 1、放回配置了preferList的Shard
        TaskItemAssignUtils.putBackWithPreferExecutorIdListConfigured(pickUpEdTaskItemList, preferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap, preferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap, preferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap);
        // 2、放回没有配置preferList的Shard
        TaskItemAssignUtils.putBackWithoutPreferExecutorIdListConfigured(pickUpEdTaskItemList, preferExecutorSettings.taskIdToOwnTrafficOnlineExecutorListMap);
    }


}
